package language.advanced.Concurrent;

import org.apache.commons.lang3.RandomUtils;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.Collectors;

//使用并行流的例子
public class parallelStreamTest {

    @Test  //并行流+ForkJoinPool指定执行线程数
    public void fjpoll(){
        ForkJoinPool forkJoinPool = new ForkJoinPool(5);
//        ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors() << 3);
//        IO阻塞 n*2

        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);

        ForkJoinTask<List<Integer>> submit = forkJoinPool.submit(() ->
                numbers.parallelStream()
                        .map(x -> addOp(x, forkJoinPool))
                        .collect(Collectors.toList())
        );
        List<Integer> join = submit.join();
        System.out.println(join);
    }

    private Integer addOp(int x, ForkJoinPool forkJoinPool) {
        Thread t = Thread.currentThread();
        String name = t.getName();
        int stime = RandomUtils.nextInt(200, 1000);
        System.out.println("示例1 Thread:" + name + " value:" + x + " sleep time " + stime + " activeCount :" + forkJoinPool.getActiveThreadCount() + " queueSize " + forkJoinPool.getQueuedTaskCount());
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return x + 1;
    }


    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        numbers.parallelStream().forEach(num -> System.out.println(num));

        List<String> strs = Arrays.asList("1", "2", "3", "1", "2", "3", "1", "2", "3", "1", "2", "3");

        strs.parallelStream()
                .forEach(str->{
                    Thread thread = new MyThread1("world");
                    thread.start();
                });



        List<String> collect = strs.parallelStream()
                .map(str -> s(str)).collect(Collectors.toList());
        System.out.println(collect);

        //如果有个一个发生异常，就直接报错
        List<Double> collectDouble = numbers.parallelStream()
                .map(num -> e(num)).collect(Collectors.toList());
        System.out.println(collectDouble);

    }

    public static double e(int a){
        return 1/a;
    }

    public static String s(String name){
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("name:"+name);
        return name+"Enum";
    }

}

class MyThread1 extends Thread {
    private String name;

    public MyThread1(String name) {
        this.name = name;
    }

    public void run() {

        System.out.println("hello " + name);
        try {
            //先等待一段时间，bQueue.add("a")模拟生产者
            Thread.sleep(8000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        Thread thread = new MyThread1("world");
        thread.start();
    }
}
